package com.itcam.kafkastream.sample;

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.apache.kafka.streams.kstream.ValueMapper;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

/**
 * @author : Cammy.Wu
 * Description : 
 */

public class KafkaStreamQuickStart {

    public static void main(String[] args) {
        // kafka的配置信息
        Properties prop = new Properties();
        prop.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.200.130:9092");
        prop.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        prop.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        prop.put(StreamsConfig.APPLICATION_ID_CONFIG, "stream-quickstart");

        // stream构建其
        StreamsBuilder streamsBuilder = new StreamsBuilder();

        // 流式计算
        streamProcessor(streamsBuilder);

        // 创建kafkaStream对象
        KafkaStreams kafkaStreams = new KafkaStreams(streamsBuilder.build(), prop);

        // 开启流式计算
        kafkaStreams.start();
    }

    /**
     * 流式计算
     * @param streamsBuilder
     */
    private static void streamProcessor(StreamsBuilder streamsBuilder) {
        // 创建KStream对象，同时指定从哪个topic接收消息
        KStream<String, String> stream = streamsBuilder.stream("itcam-topic-input");
        /**
         * 处理消息的value
         */
        stream.flatMapValues(new ValueMapper<String, Iterable<String>>() {
                    @Override
                    public Iterable<String> apply(String value) {
                        String[] valAry = value.split(" ");
                        return Arrays.asList(valAry);
                    }
                })
                // 按照value进行聚合处理
                .groupBy((key, value) -> value)
                /*
                    时间窗口在Kafka Streams中的作用可以用一个生活中的例子来解释。
                    假设你是一个咖啡店的老板，你想了解在一天中的不同时间段，顾客对不同种类咖啡的偏好。你可以使用时间窗口来分析这个问题。
                    数据聚合：你可以将一天分成若干个时间段，比如每小时一个窗口。然后，你可以统计每个窗口内卖出的咖啡种类和数量。这样，你就可以看到在不同时间段，哪种咖啡最受欢迎。
                    事件时间处理：在这个例子中，事件时间就是顾客购买咖啡的时间。即使数据处理可能会有延迟，时间窗口也能确保你基于实际购买时间来分析数据，而不是数据被处理的时间。
                    实时分析：通过时间窗口，你可以实时监控咖啡店的销售情况，了解顾客的偏好是否随着时间的变化而变化。
                    减少数据量：将数据按照时间窗口进行聚合，可以减少需要处理的数据量，提高处理效率和性能。你不需要分析每一笔交易，只需要关注每个时间段的汇总数据。
                    在Kafka Streams中，时间窗口通常与TimeWindows类一起使用，它提供了不同类型的窗口，如滚动窗口（Tumbling Windows）、
                        滑动窗口（Sliding Windows）和会话窗口（Session Windows）。每种窗口类型都有其特定的应用场景和使用方法。
                    例如，如果你想要统计每小时内的咖啡销售总量，你可以使用滚动窗口。如果你想要分析顾客在一天中的不同时间段的购买行为模式，
                        你可以使用滑动窗口。如果你想要了解顾客在店内停留期间的购买习惯，你可以使用会话窗口。
                    总之，时间窗口是Kafka Streams中非常重要的一个概念，它为流式数据处理提供了强大的时间维度分析能力，帮助你更好地理解和利用数据
                 */
                // 时间窗口
                .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))
                // 统计单词的个数
                .count()
                // 转换为kStream对象
                .toStream()
                // 对聚合处理后的结果进行映射
                .map((key, value) -> {
                    // 打印每个键值对
                    System.out.println("key:" + key.key() + " value:" + value);
                    // 将键值对转换为新的键值对
                    return new KeyValue<>(key.key().toString(), value.toString());
                })
                // 发送消息
                .to("itcam-topic-out");
    }
}
